package org.example.SparkStreamCount

import org.apache.spark.sql.{SparkSession, SaveMode}
import org.apache.spark.sql.functions._

object SaveToMySQL {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("SaveToMySQL")
      .master("yarn") // 或者 local
      .getOrCreate()

    // 读取统计数据
    val outputPath = "hdfs://192.168.60.128:50070/explorer.html#/cxx/mytest/part-00000/"
    val dataToSaveRdd = spark.read.textFile(outputPath).rdd.map { line =>
      val parts = line.split("\t") // 假设以制表符分隔
      (parts(0), parts(1).toInt) // (单词, 频率)
    }

    // 将 RDD 转换为 DataFrame
    import spark.implicits._
    val dataToSave = dataToSaveRdd.toDF("word", "count")

    // 数据验证：检查有效性
    val validatedData = dataToSave.filter(col("word").isNotNull && trim(col("word")) =!= "" && col("count") >= 0)

    // 记录无效的行数
    val invalidCount = dataToSave.count() - validatedData.count()
    if (invalidCount > 0) {
      println(s"Warning: Found $invalidCount invalid records. These records will be skipped.")
    }

    // 保存有效数据到数据库
    validatedData.write
      .mode(SaveMode.Append)
      .jdbc("jdbc:mysql://localhost:3306/MySparkCount", "cjy_wordcount", new java.util.Properties() {{
        put("user", "root")
        put("password", "123")
      }})

    // 关闭 SparkSession
    spark.stop()
  }
}